package cn.doitedu.flinkapi.apidemos;

import cn.doitedu.flinkapi.commons.DetectorTemp;
import cn.doitedu.flinkapi.commons.RandomTempSource;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class TransformDemo {

    public static void main(String[] args) throws Exception {
        ExecutionConfig config = new ExecutionConfig();
        config.setAutoWatermarkInterval(200);

        Configuration conf = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        //env.setParallelism(1);
        WatermarkStrategy<String> timestampAssigner = WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofMillis(1000)).withTimestampAssigner(new SerializableTimestampAssigner<String>() {
            @Override
            public long extractTimestamp(String element, long recordTimestamp) {

                return Long.parseLong(element.split(",")[1]);
            }
        });


        /**
         * source
         */
        SourceFunction<String> sf;
        DataStream<String> source = env
                .addSource(new RandomTempSource(2000),"randomsource", TypeInformation.of(new TypeHint<String>() {}))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofMillis(2000)).withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                    @Override
                    public long extractTimestamp(String element, long recordTimestamp) {

                        return 0;
                    }
                }))

                .assignTimestampsAndWatermarks(timestampAssigner)
                .setParallelism(1);
        //source.print("source");

        /**
         * map
         */
        DataStream<DetectorTemp> map = source.map(s -> {
            String[] split = s.split(",");
            DetectorTemp detectorTemp = new DetectorTemp(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2]), Double.parseDouble(split[3]));
            return detectorTemp;
        }).setParallelism(2);
        //map.print("map");

        /**
         * flatmap
         */
        SingleOutputStreamOperator<Tuple3<String, Long, Double>> flatMap = source.flatMap((v, o) -> {
            String[] split = v.split(",");
            o.collect(new Tuple3<String, Long, Double>(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2])));
            o.collect(new Tuple3<String, Long, Double>(split[0], Long.parseLong(split[1]), Double.parseDouble(split[3])));
        }, TypeInformation.of(new TypeHint<Tuple3<String, Long, Double>>() {
        }));
        //flatMap.print("flatmap");

        /**
         * relalance
         * shuffle
         */
        DataStream<DetectorTemp> rebalance = map.rebalance();
        DataStream<DetectorTemp> shuffle = map.shuffle();
        rebalance.print("reb").setParallelism(2);
        shuffle.print("shuffle").setParallelism(2);


        /**
         * side output
         */
        OutputTag<String> sideTag = new OutputTag<String>("side",TypeInformation.of(String.class));
        SingleOutputStreamOperator<String> side = source.process(new ProcessFunction<String, String>() {

            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                out.collect(value);
                ctx.output(sideTag, value.split(",")[0]);
            }
        });

        side.print("side");
        DataStream<String> side1 = side.getSideOutput(sideTag);
        side1.print("side1");


        /**
         * connect
         */
        ConnectedStreams<String, String> connect = side.connect(side1);
        SingleOutputStreamOperator<String> map1 = connect.map(new CoMapFunction<String, String, String>() {
            @Override
            public String map1(String value) throws Exception {
                return value;
            }

            @Override
            public String map2(String value) throws Exception {
                return value + "_side1";
            }
        });
        map1.print("connect");


        env.execute();

    }
}
